package com.tinghua.mqTest.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;
import com.tinghua.mqTest.utils.ConnectionMQ;

import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * 发布确认
 */
public class ProducerConfirm {
    public static final String QUEUE_NAME = "rabbitmq_01_test";
    public static final boolean DURABLE = true;
    public static final int MESSAGE_COUNT = 1000;
    public static void main(String[] args) throws Exception {
        //单个发布确认
        //OneConfirm();        //耗费时长:25636
        //批量发布确认
        //batchConfirm();      //耗费时长:319
        //异步发布确认
        asynConfirm();         //耗费时长:47
    }

    /**
     * 该方法时间耗费长
     */
    public static void OneConfirm() throws IOException, TimeoutException, InterruptedException {
        Channel channel = ConnectionMQ.utilConnectionMQ();
        channel.queueDeclare(QUEUE_NAME, DURABLE, false, false, null);
        channel.confirmSelect();
        long beginTime = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, String.valueOf(i).getBytes());
            channel.waitForConfirms();
        }
        long finalTime = System.currentTimeMillis() - beginTime;
        System.out.println("耗费时长:" + finalTime);
    }

    /**
     * 该方法不安全，不能保证数据安全
     */
    public static void batchConfirm() throws Exception{
        Channel channel = ConnectionMQ.utilConnectionMQ();
        channel.queueDeclare(QUEUE_NAME, DURABLE, false, false, null);
        channel.confirmSelect();
        long beginTime = System.currentTimeMillis();
        int confirmCount = 0;
        for (int i = 0; i < MESSAGE_COUNT; i++,confirmCount++) {
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, String.valueOf(i).getBytes());
            if (i % 100 == 0){
                channel.waitForConfirms();
            }
        }
        //确保没有剩余未被确认的消息
        if (confirmCount % 100 != 0){
            channel.waitForConfirms();
        }
        long finalTime = System.currentTimeMillis() - beginTime;
        System.out.println("耗费时长:" + finalTime);
    }

    /**
     * 该方法是最高效的
     * @throws Exception
     */
    public static void asynConfirm() throws Exception{
        Channel channel = ConnectionMQ.utilConnectionMQ();
        channel.queueDeclare(QUEUE_NAME, DURABLE, false, false, null);
        channel.confirmSelect();
        ConcurrentSkipListMap<Long, String> map = new ConcurrentSkipListMap<>();
        //成功的回调
        ConfirmCallback confirmCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long, String> navigableMap = map.headMap(deliveryTag);
                navigableMap.clear();
            } else {
                map.remove(deliveryTag);
            }
            System.out.println("确认成功的消息" + deliveryTag);

        };
        ConfirmCallback noConfirmCallback = (deliveryTag, multiple) -> {
            System.out.println(map.get(deliveryTag).toString());
            System.out.println("确认失败的消息" + deliveryTag);
        };
        //开启监听，由远程rabbitMQ告诉我们是否存储成功
        channel.addConfirmListener(confirmCallback, noConfirmCallback);
        long beginTime = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, String.valueOf(i).getBytes());
            map.put(channel.getNextPublishSeqNo(), String.valueOf(i));
        }
        long finalTime = System.currentTimeMillis() - beginTime;
        System.out.println("耗费时长:" + finalTime);
    }
}
